package com.aotain.jupiter.abnormal;

import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;

import org.apache.commons.codec.binary.Base64;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.tuple.Tuple2;

public class AbnormalReduce implements ReduceFunction<Tuple2<String,AbnormalStatTuple>>{

	/** 
	* @Fields serialVersionUID : TODO(用一句话描述这个变量表示什么) 
	*/ 
	private static final long serialVersionUID = -6119926283638667302L;
	

	private HashMap<String,Long> sipPVMap = new HashMap<String,Long>();//目标IP PV数
	private HashMap<String,Long> sdPVMap = new HashMap<String,Long>();//各个源ip到目标ip的PV数
	private Map<String,Long> tripleMap = new HashMap<String,Long>();//各个元组的PV
	private Map<String,Set<String>> dipToTriple = new HashMap<String,Set<String>>();//目标ip下的三元组映射
	private Map<String,Long> delayTimeMap = new HashMap<String,Long>();//求延时Map
	private Map<String,Long> maxDelayMap = new HashMap<String,Long>();//最大延时Map
	private Map<String,Long> minDelayMap = new HashMap<String,Long>();//最小延时Map
	private Map<String,Long> streamPacketMap = new HashMap<String,Long>();//
	private Map<String,Long> delayTimePVMap = new HashMap<String,Long>();//R

	@Override
	public Tuple2<String, AbnormalStatTuple> reduce(
			Tuple2<String, AbnormalStatTuple> value1,
			Tuple2<String, AbnormalStatTuple> value2) throws Exception {
		
		if(value1 == null || value2 == null)
			return null; 
		Tuple2<String,AbnormalStatTuple> retTuple = new Tuple2<String,AbnormalStatTuple>();
		
//		Logger.getLogger(getClass()).info("#####AbnormalStatMap#### tuple is " + value.f1);
		String sip1 = value1.f1.getSourceIP();
		String dip1 = value1.f1.getDestIP();
		String gis1 = new String(value1.f1.getGIS());
		String idc1 = value1.f1.getIDC();
		String type1 = value1.f1.getType();
		String dport1 = value1.f1.getDestPort();
		String sport1 = value1.f1.getSourcePort();
		long delayTime1 = value1.f1.getDelayTime();
		long upStreamOctets1 = value1.f1.getUpStreamOctets();
		long upStreamPacket1 = value1.f1.getUpStreamPacket();
		
		String tripleKey = dip1 + "*" + idc1 + "*" + sip1 + "*" + gis1 + "*" + type1 + "*" + dport1;
		String idc_sip = value1.f0;
		String sdipKey = dip1 + "*" + idc1 + "*" + sip1;
		AbnormalStatTuple stat = new AbnormalStatTuple();
		stat.setSourceIP(sip1);
		stat.setDestIP(dip1);
		stat.setGIS(gis1);
		stat.setType(type1);
		stat.setDestPort(dport1);
		stat.setSourcePort(sport1);
		stat.setDelayTime(delayTime1);
		stat.setUpStreamOctets(upStreamOctets1);
		stat.setUpStreamPacket(upStreamPacket1);
		
		
		long dipPV = 0l;
		long triplePV = upStreamOctets1;
		long sdPV = 0;
		long triplePacket = upStreamPacket1;
		long delayPV = 1;
		
		if(sipPVMap.get(idc_sip) != null) {
			dipPV = sipPVMap.get(idc_sip) + upStreamOctets1;
		}
		if(tripleMap.get(tripleKey) != null) {
			triplePV = tripleMap.get(tripleKey) + upStreamOctets1;
		}
		if(sdPVMap.get(sdipKey) != null) {
			sdPV = sdPVMap.get(sdipKey) + upStreamOctets1;
		}
		if(streamPacketMap.get(tripleKey) != null) {
			triplePacket = streamPacketMap.get(tripleKey) + upStreamPacket1;
		}
		if(delayTimePVMap.get(sdipKey) != null) {
			delayPV = delayTimePVMap.get(sdipKey) + 1;
		}
		if(maxDelayMap.get(sdipKey) == null) {
			maxDelayMap.put(sdipKey, delayTime1);
		}
		if(delayTime1>maxDelayMap.get(sdipKey)) {
			maxDelayMap.put(sdipKey, delayTime1);
		}
		if(minDelayMap.get(sdipKey) == null) {
			minDelayMap.put(sdipKey, delayTime1);
		}
		if(delayTime1<minDelayMap.get(sdipKey)) {
			minDelayMap.put(sdipKey, delayTime1);
		}
		if(delayTimeMap.get(sdipKey) != null) {
			delayTime1 = delayTimeMap.get(sdipKey) + delayTime1;
		}

		Set<String> tripleSet = dipToTriple.get(idc_sip);
		if(tripleSet == null) {
			tripleSet = new HashSet<String>();
		}
		tripleSet.add(tripleKey);
		dipToTriple.put(idc_sip, tripleSet);

		sdPVMap.put(sdipKey, sdPV);
		sipPVMap.put(idc_sip, dipPV);
		tripleMap.put(tripleKey, triplePV);
		delayTimeMap.put(sdipKey, delayTime1);
		streamPacketMap.put(tripleKey, triplePacket);
		delayTimePVMap.put(sdipKey, delayPV);
		
		stat.setDipToTriple(dipToTriple);
		stat.setSdPVMap(sdPVMap);
		stat.setSipPVMap(sipPVMap);
		stat.setTripleMap(tripleMap);
		stat.setStreamPacketMap(streamPacketMap);
		stat.setDelaytimeMap(delayTimeMap);
		stat.setDelayTimePVMap(delayTimePVMap);
		
		String sip2 = value2.f1.getSourceIP();
		String dip2 = value2.f1.getDestIP();
		String gis2 = new String(value2.f1.getGIS());
		String idc2 = value2.f1.getIDC();
		String type2 = value2.f1.getType();
		String dport2 = value2.f1.getDestPort();
		String sport2 = value2.f1.getSourcePort();
		long delayTime2 = value2.f1.getDelayTime();
		long upStreamOctets2 = value2.f1.getUpStreamOctets();
		long upStreamPacket2 = value2.f1.getUpStreamPacket();
		
		String tripleKey2 = dip2 + "*" + idc2 + "*" + sip2 + "*" + gis2 + "*" + type2 + "*" + dport2;
		String sdipKey2 = dip2 + "*" + idc2 + "*" + sip2;
//		String key = dip + "*" + idc;
		stat.setSourceIP(sip2);
		stat.setDestIP(dip2);
		stat.setGIS(gis2);
		stat.setType(type2);
		stat.setDestPort(dport2);
		stat.setSourcePort(sport2);
		stat.setDelayTime(delayTime2);
		stat.setUpStreamOctets(upStreamOctets2);
		stat.setUpStreamPacket(upStreamPacket2);
		
		long dipPV2 = 0l;
		long triplePV2 = upStreamOctets2;
		long sdPV2 = 0;
		long triplePacket2 = upStreamPacket2;
		long delayPV2 = 1;
		
		if(sipPVMap.get(idc_sip) != null) {
			dipPV2 = sipPVMap.get(idc_sip) + upStreamOctets2;
		}
		if(tripleMap.get(tripleKey2) != null) {
			triplePV2 = tripleMap.get(tripleKey2) + upStreamOctets2;
		}
		if(sdPVMap.get(sdipKey2) != null) {
			sdPV2 = sdPVMap.get(sdipKey2) + upStreamOctets2;
		}
		if(streamPacketMap.get(tripleKey2) != null) {
			triplePacket2 = streamPacketMap.get(tripleKey2) + upStreamPacket2;
		}
		if(delayTimePVMap.get(sdipKey2) != null) {
			delayPV2 = delayTimePVMap.get(sdipKey2) + 1;
		}
		if(maxDelayMap.get(sdipKey2) == null) {
			maxDelayMap.put(sdipKey2, delayTime2);
		}
		if(delayTime2>maxDelayMap.get(sdipKey2)) {
			maxDelayMap.put(sdipKey2, delayTime2);
		}
		if(minDelayMap.get(sdipKey2) == null) {
			minDelayMap.put(sdipKey2, delayTime2);
		}
		if(delayTime2<minDelayMap.get(sdipKey2)) {
			minDelayMap.put(sdipKey2, delayTime2);
		}
		if(delayTimeMap.get(sdipKey2) != null) {
			delayTime2 = delayTimeMap.get(sdipKey2) + delayTime2;
		}

		tripleSet.add(tripleKey2);
		dipToTriple.put(idc_sip, tripleSet);

		sdPVMap.put(sdipKey2, sdPV2);
		sipPVMap.put(idc_sip, dipPV2);
		tripleMap.put(tripleKey2, triplePV2);
		delayTimeMap.put(sdipKey2, delayTime2);
		streamPacketMap.put(tripleKey2, triplePacket2);
		delayTimePVMap.put(sdipKey2, delayPV2);
		
		stat.setDipToTriple(dipToTriple);
		stat.setSdPVMap(sdPVMap);
		stat.setSipPVMap(sipPVMap);
		stat.setTripleMap(tripleMap);
		stat.setStreamPacketMap(streamPacketMap);
		stat.setDelaytimeMap(delayTimeMap);
		stat.setDelayTimePVMap(delayTimePVMap);
		
		retTuple.setFields(idc_sip, stat);
		return retTuple;
	}

}
